package Fp

import (
	"fmt"
	"db2s/parDef"
	"reflect"
	"strings"
	"sync/atomic"
	"time"
)

func (sp TaskDistribution) getSubLeftDirection(s *atomic.Value) (res []map[string]string) {
	if s == nil {
		return
	}
	return getAtomicValue(s).([]map[string]string)
}
func (sp TaskDistribution) inputConditionNullIf(s parDef.IndexPart) bool {
	left := sp.getSubLeftDirection(s.WhereGenerate.LeftDirection)
	g := left[len(left)-1]
	h, hok := g["columnDataLeft"]
	if !hok {
		return false
	}
	if strings.EqualFold(h, "NULL") {
		return true
	}
	return false
}
func (sp TaskDistribution) inputLeftEqNullActive(s parDef.IndexPart) { // == null
	var (
		event = "[second-inputEqNullActive]"
	)
	sp.subSendMsg(s)
	infoActive(fmt.Sprintf("(%v-%v) %v setupEnd&: send NULL index segment:left:{%v} right:{%v} of table %v.%v(%v).", s.LogSeq, sp.LogSeq, event, getAtomicValue(s.WhereGenerate.LeftDirection), getAtomicValue(s.WhereGenerate.RightDirection), sp.TableObject.Schema, sp.TableObject.Table, sp.TableObject.PartitionName))
}
func (sp TaskDistribution) inputLeftNullEqGtSplit(left []map[string]string) (null, notNull []map[string]string) {
	var (
		eqSymbol, gtSymbol = make(map[string]string), make(map[string]string)
	)
	null, notNull = append(null, left[:len(left)-1]...), append(notNull, left[:len(left)-1]...)
	g := left[len(left)-1]
	for k, v := range g {
		if strings.EqualFold(k, "querySymbolLeft") {
			eqSymbol[k] = "="
			gtSymbol[k] = ">"
		} else {
			eqSymbol[k] = v
			gtSymbol[k] = v
		}
	}

	null = append(null, eqSymbol)
	notNull = append(notNull, gtSymbol)
	return
}
func (sp TaskDistribution) inputRightGtEqSplit(left, right []map[string]string) []map[string]string {
	var z []map[string]string
	for _, v := range right {
		if !strings.EqualFold(v["columnName"], left[len(left)-1]["columnName"]) {
			z = append(z, v)
		}
	}
	return z
}
func (sp TaskDistribution) inputGtEqNullActive(s parDef.IndexPart) { //>= null
	var (
		event         = "[second-inputGtEqNullActive]"
		null, notNull []map[string]string
	)
	left := sp.getSubLeftDirection(s.WhereGenerate.LeftDirection)
	right := sp.getSubLeftDirection(s.WhereGenerate.RightDirection)
	null, notNull = sp.inputLeftNullEqGtSplit(left)
	newRight := sp.inputRightGtEqSplit(left, right)
	c := sp.pInputInit()
	swapAtomicValue(c.WhereGenerate.LeftDirection, null)
	swapAtomicValue(c.WhereGenerate.RightDirection, newRight)
	infoActive(fmt.Sprintf("(%v-%v) %v setupEnd& send >= NULL index segment:left:{%v} right:{%v} of table %v.%v(%v).", s.LogSeq, sp.LogSeq, event, getAtomicValue(c.WhereGenerate.LeftDirection), getAtomicValue(c.WhereGenerate.RightDirection), sp.TableObject.Schema, sp.TableObject.Table, sp.TableObject.PartitionName))
	sp.subSendMsg(c)
	swapAtomicValue(s.WhereGenerate.LeftDirection, notNull)
}
func (sp TaskDistribution) inputConditionNullSplit(s parDef.IndexPart) {
	left := sp.getSubLeftDirection(s.WhereGenerate.LeftDirection)
	g := left[len(left)-1]
	i, iok := g["querySymbolLeft"]
	if !iok {
		return
	}
	switch i {
	case "=":
		sp.inputLeftEqNullActive(s)
	case ">=":
		sp.inputGtEqNullActive(s)
	}
}
func (sp TaskDistribution) inputDirectSendMsgIf(s parDef.IndexPart) bool {
	//var event = "[Second]"
	if s.Limit.Pagination != nil && atomic.LoadInt64(s.Limit.Pagination) <= 300 {
		return true
	}

	//if atomic.LoadInt64(s.Limit.Pagination) > 300 {
	//	return true
	//}
	return false
}
func (sp TaskDistribution) subProtectionMechanism() {
	var (
		event = "[protectionMechanism]"
	)
	md5Str := fmt.Sprintf("table: %s.%v(%v) sq curry length: %v wherestart:%v wherecolumn:%v limit %v", sp.TableObject.Schema, sp.TableObject.Table, sp.TableObject.PartitionName, sp.StaticP.CpLength, sp.subGetWsValue(), sp.subGetWcValue(), atomic.LoadInt64(sp.Second.Batch.Limit.Pagination))
	if m5 := md5Hash(md5Str); fmt.Sprintf("%v", sp.subGetSwsMd5()) == m5 {
		time.Sleep(1 * time.Second)
		swapLoopFrequency(sp.Second.LoopFrequency, atomic.LoadInt64(sp.Second.LoopFrequency)+1)
		if atomic.LoadInt64(sp.Second.LoopFrequency) > LoopThreshold {
			errorActive(fmt.Sprintf("(%d-%v) %s setup3&: An death loop occurs in table %v.%v(%v), and the table will stop operating. md5Str info %v", sp.LogSeq, sp.getFirstSemLogSeq(), event, sp.TableObject.Schema, sp.TableObject.Table, sp.TableObject.PartitionName, md5Str))
			swapErrorPoint(sp.ErrEnd)
			return
		}
	} else {
		swapLoopFrequency(sp.Second.LoopFrequency, 1)
		swapAtomicValue(sp.Second.BreakMd5, m5)
	}
}
func (sp TaskDistribution) subResultGroup(s parDef.IndexPart, res1 whereLeft, res2 []whereRight) (p []parDef.IndexPart) {
	var (
		event = "[resultGroup]"
	)
	right := getRightDirection(s.WhereGenerate.RightDirection)
	if len(res1.swsLeft) > 0 {
		swapAtomicValue(sp.Second.Batch.WhereGenerate.LeftDirection, res1.swsLeft)
	}
	swapAtomicValue(sp.Second.Batch.WhereGenerate.RightDirection, getAtomicValue(s.WhereGenerate.RightDirection))
	if len(res1.swsRight) < len(right) && len(res1.swsRight) > 0 && len(res1.swsLeft) > 0 { //生成的sws 右边条件数量小于一层传入的，即现有范围大于传入的
		swapAtomicValue(sp.Second.Batch.WhereGenerate.RightDirection, right)
	}
	for _, x := range res2 {
		y := sp.pOutputInit()
		y.WhereGenerate.IndexColumn = sp.getSecondNameStringSlice()
		swapAtomicValue(y.WhereGenerate.LeftDirection, x.pLeft)
		swapAtomicValue(y.WhereGenerate.RightDirection, x.pRight)
		y.PassP.Ws = sp.subGetWsValue()
		y.PassP.Wc = sp.subGetWcValue()
		y.LogSeq = sp.LogSeq
		y.Limit.Segment = x.Segment
		atomic.SwapInt64(y.Limit.Pagination, atomic.LoadInt64(sp.Second.Batch.Limit.Pagination))
		//atomic.SwapInt32(y.WhereGenerate.NullBool, atomic.LoadInt32(x.NullBool))
		infoActive(fmt.Sprintf("(%d-%v)  %v setup5&: Index segment general successful of table%v.%v(%v) limit[default] limitReset:%v general:{left:%v right:%v}.", sp.LogSeq, sp.LogSeq, event, sp.TableObject.Schema, sp.TableObject.Table, sp.TableObject.PartitionName, y.Limit.Segment, getAtomicValue(sp.Second.Batch.WhereGenerate.LeftDirection), getAtomicValue(sp.Second.Batch.WhereGenerate.RightDirection)))
		p = append(p, *y)
	}
	return
}
func (sp TaskDistribution) subConditionalValue(s inputConditionalS) (res1 whereLeft, res2 []whereRight, err error) {
	var site string
	if site, err = sp.getTwoValueCompare(s); err != nil {
		return
	}
	if fun, ok := sp.getConditionalFuncReflectValue()[site]; ok {
		args := []reflect.Value{
			reflect.ValueOf(valueLocationInput{
				k:        atomic.LoadInt64(s.CompareSeq),
				ws:       sp.subGetWsValue(),
				wc:       sp.subGetWcValue(),
				indexCol: sp.getSecondNameStringSlice(),
				//nullMark: getNullValue(s.Null),
			}),
		}
		q := fun.Call(args)
		res1 = q[0].Interface().(whereLeft)
		res2 = q[1].Interface().([]whereRight)
	}
	return
}
func (sp TaskDistribution) subLimitSegmentationValue(s parDef.IndexPart) (p []parDef.IndexPart, err error) {
	var (
		compareSeq int64
		res1       whereLeft
		res2       []whereRight
	)
	if res1, res2, err = sp.subConditionalValue(inputConditionalS{
		full:       sp.Full,
		Name:       sp.Second.Name,
		CompareSeq: &compareSeq,
		Type:       sp.Second.Type,
		Null:       sp.Second.Null,
		WS:         sp.Second.WS,
		WC:         sp.Second.WC,
	}); err != nil {
		return
	}
	p = sp.subResultGroup(s, res1, res2)
	switch atomic.LoadInt64(sp.Second.Batch.Limit.Pagination) {
	case 0:
		switch len(sp.subGetWsValue()) {
		case 0: //first line
		default: //end line or two line
			if atomic.LoadInt32(sp.Second.Batch.Limit.ModifyS) == 0 {
				swapLimitPagination(sp.Second.Batch.Limit.Pagination, *sp.StaticP.MulFactor*sp.StaticP.ChunkNumber)
			}
		}
	case 1:
		swapLimitPagination(sp.Second.Batch.Limit.Pagination, 0)
	default:
		if atomic.LoadInt32(sp.Second.Batch.Limit.ModifyS) == 0 {
			swapLimitPagination(sp.Second.Batch.Limit.Pagination, *sp.StaticP.MulFactor*sp.StaticP.ChunkNumber)
		}
	}
	swapAtomicValue(sp.Second.WS, sp.subGetWcValue())
	return
}
func (sp TaskDistribution) subOutputResult(s parDef.IndexPart) (err error) {
	var (
		event = "[Second]"
		res   []parDef.IndexPart
	)
	if res, err = sp.subLimitSegmentationValue(s); err != nil {
		return err
	}
	for _, l := range res {
		if len(getLeftDirection(l.WhereGenerate.LeftDirection)) > 0 && len(getLeftDirection(l.WhereGenerate.RightDirection)) > 0 {
			infoActive(fmt.Sprintf("(%d-%v) %v setupEnd&: send Msg Index segment of table:%v.%v(%v). general sql: left %v right %v", sp.LogSeq, sp.LogSeq, event, sp.TableObject.Schema, sp.TableObject.Table, sp.TableObject.PartitionName, getAtomicValue(l.WhereGenerate.LeftDirection), getAtomicValue(l.WhereGenerate.RightDirection)))
			l.WhereGenerate.NewLeftDirection = getLeftDirection(l.WhereGenerate.LeftDirection)
			l.WhereGenerate.NewRightDirection = getLeftDirection(l.WhereGenerate.RightDirection)
			sp.subSendMsg(l)
		}
	}
	return
}

func (sp TaskDistribution) IsConcurrentValue(s parDef.IndexPart) {
	var (
		bf    = false
		event = "[Second]"
		vlog  string
	)
	infoActive(fmt.Sprintf("(%v-%v) %v setup1&: Start the second level parallel query segmentation of table %v.%v(%v).", s.LogSeq, sp.LogSeq, event, sp.TableObject.Schema, sp.TableObject.Table, sp.TableObject.PartitionName))
	if sp.inputConditionNullIf(s) { //存在null值
		sp.inputConditionNullSplit(s)
		sp.subCloseMsg()
		return
	}
	if sp.inputDirectSendMsgIf(s) { //直接发送
		sp.subSendMsg(s)
		sp.subCloseMsg()
		return
	}
	if !sp.subRunInit(s) {
		sp.subCloseMsg()
		return
	}
	go func() {
	breakOut:
		for {
			if sp.subErrClose(s.LogSeq, sp.LogSeq) {
				break breakOut
			}
			sp.Second.Batch.NewIndexCol = sp.getSecondNameStringSlice()
			if err := sp.getSegmentationData(sp.Second.Batch, sp.Second.WC); err != nil {
				break breakOut
			}
			sp.subProtectionMechanism()
			vlog = fmt.Sprintf("(%v-%v) %v setup3&: Index segment query successful of table:%v.%v(%v). indexColumn:%v,sws:{left:%v right:%v ls:%v,ws:%v,wc:%v,breakf:%v}", s.LogSeq, sp.LogSeq, event, sp.TableObject.Schema, sp.TableObject.Table, sp.TableObject.PartitionName, sp.Second.Batch.WhereGenerate.IndexColumn, getAtomicValue(sp.Second.Batch.WhereGenerate.LeftDirection), getAtomicValue(sp.Second.Batch.WhereGenerate.RightDirection), atomic.LoadInt64(sp.Second.Batch.Limit.Pagination), sp.subGetWsValue(), sp.subGetWcValue(), bf)
			infoActive(vlog)
			if res, err := sp.subNotSegmentationValue(s); err != nil {
				swapErrorPoint(sp.ErrEnd)
				break breakOut
			} else {
				switch res {
				case 1:
					continue
				case 2:
					break breakOut
				}
			}
			if sp.subOutputResult(s) != nil {
				swapErrorPoint(sp.ErrEnd)
				break breakOut
			}
		}
		sp.subCloseMsg()
	}()
	return
}
